package com.mercury.rabbitmq.manager;

import com.alibaba.fastjson.JSONObject;
import com.mercury.rabbitmq.domain.RabbitMqMessage;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * 消息队列-生产者 业务层处理
 *
 * @author mercury
 */
@Component
public class RabbitMqManager {


    @Resource
    private RabbitTemplate rabbitTemplate;

    @Resource
    private RedisTemplate<String, String> redisTemplate;


    /**
     * 发送消息
     *
     * @param content    消息内容
     * @param exchange   交换机
     * @param routingKey 路由key
     */
    public void sendMsg(String exchange, String routingKey, RabbitMqMessage content) {
        Message message = buildMessage(content);
        CorrelationData data = new CorrelationData(content.getId());
        //存储到redis
        redisTemplate.opsForValue().set(data.getId(), JSONObject.toJSONString(content));
        rabbitTemplate.convertAndSend(exchange, routingKey, message, data);
    }


    /**
     * 延迟发送消息(使用延时插件)
     * <p>
     * 大概流程将消息丢到延时交换机,延时时间到了在投递到队列中
     *
     * @param exchange    延迟交换机
     * @param routingKey  延迟路由key
     * @param content     延迟发送的实体对象时,实体对象implements Serializable
     * @param delayMillis 延迟时间,单位 毫秒
     */
    public void sendDelayedMsg(String exchange, String routingKey, Integer delayMillis, RabbitMqMessage content) {
        Message message = buildMessage(content);
        //自定义rabbitmq id
        CorrelationData data = new CorrelationData(content.getId());
        redisTemplate.opsForValue().set(data.getId(), JSONObject.toJSONString(content));
        rabbitTemplate.convertAndSend(exchange, routingKey, message, msg -> {
            // 设置延迟毫秒值=delayTime*1000
            msg.getMessageProperties().setDelay(delayMillis);
            return msg;
        }, data);
    }


    /**
     * 延迟发送消息(死信队列)
     * <p>
     * 整体流程就是发送一条消息到普通队列但是消息设置过期时间,等过期时间到了自动投递到死信交换机,消费
     *
     * @param exchange   业务交换机
     * @param routingKey 业务路由key
     * @param content    业务发送的实体对象时,实体对象implements Serializable
     * @param expiration 过期时间,单位 毫秒
     */
    public void sendDeadLetterMsg(String exchange, String routingKey, Integer expiration, RabbitMqMessage content) {
        Message message = buildMessage(content);
        //自定义rabbitmq id
        CorrelationData data = new CorrelationData(content.getId());
        redisTemplate.opsForValue().set(data.getId(), JSONObject.toJSONString(content));
        rabbitTemplate.convertAndSend(exchange, routingKey, message, msg -> {
            // 设置延迟毫秒值=expiration*1000
            msg.getMessageProperties().setExpiration(String.valueOf(expiration));
            return msg;
        }, data);
    }


    /**
     * 构建消息内容
     *
     * @param content 内容
     * @return 消息体
     */
    public Message buildMessage(RabbitMqMessage content) {
        return MessageBuilder
                .withBody(JSONObject.toJSONString(content).getBytes())
                .setContentType(MessageProperties.CONTENT_TYPE_JSON)
                .setCorrelationId(content.getId())
                .build();
    }


}